package com.ass.rocketmq.producer;

import com.ass.rocketmq.constant.MQTopic;
import com.ass.rocketmq.constant.ProducerGroupName;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;

/**
 * @Author: clownc
 * @Date: 2019-05-31 16:23
 */
@Log4j2
@Configuration
public class DefaultProducer {


    @Value("${spring.profiles.active}")
    private String active;

    @Value("${rocketmq.accessKey}")
    private String accessKey;

    @Value("${rocketmq.secretKey}")
    private String secretKey;

    @Value("${rocketmq.namespace}")
    private String namespace;

    @Value("${rocketmq.name-server}")
    private String namesrvAddr;




    /**
     * 创建普通消息发送者实例 全局可用 实例化一个就行 这个是 统一默认的生产组 暂时不用了
     * @return
     * @throws MQClientException
     */
    @Bean
    public DefaultMQProducer defaultProducer() throws MQClientException {
        log.info("defaultProducer 正在创建---------------------------------------");
        DefaultMQProducer producer = new DefaultMQProducer(namespace,
                ProducerGroupName.DEFAULT_GROUP_NAME.name(),
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendAsyncFailed(10);
        producer.start();
        log.info("rocketmq producer server开启成功---------------------------------.");
        return producer;
    }

    /**
     * 创建异常的生产者
     * @return
     * @throws MQClientException
     */
    @Bean
    public DefaultMQProducer errorProducer() throws MQClientException {
        log.info("errorProducer 正在创建---------------------------------------");
        DefaultMQProducer producer = new DefaultMQProducer(namespace,
                ProducerGroupName.ERROR_GROUP_NAME.name(),
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendAsyncFailed(10);
        producer.start();
        log.info("rocketmq errorProducer server开启成功---------------------------------.");
        return producer;
    }


    /**
     * 发送消息 直接配置参数在枚举里面，对应各个属性，不要写多个send方法
     * messageBody 是对应的实体转json字符串
     */
    public void send(MQTopic mqTopic, String messageBody) {
        Message message = new Message();
        message.setTopic(active + "_" + mqTopic.topic);
        message.setTags(active + "_" + mqTopic.tag);
        message.setBody(messageBody.getBytes());
        message.setKeys(UUID.randomUUID().toString());
        message.setDelayTimeLevel(mqTopic.delayTimeLevel);
        try{
            this.defaultProducer().send(message);
        }catch(Exception e){
            log.error(e);
        }
    }
}

